package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class BaseUserInfoApp {
    public static void main(String[] args) throws Exception {
        //创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //从 Kafka 读取业务数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table topic_db(" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`data` map<string, string>,\n" +
                "`ts` string\n" +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_interaction_favor_add"));
        //读取用户表数据
        Table table = tableEnv.sqlQuery("select \n" +
                "`data`['id'] as `id`,\n" +
                "`data`['name'] as `name`,\n" +
                "`data`['phone_num'] as `phone_num`,\n" +
                "`data`['email'] as `email`,\n" +
                "`data`['user_level'] as `user_level`,\n" +
                "`data`['birthday'] as `birthday`,\n" +
                " TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) as `age`,\n" +
                "`data`['gender'] as `gender`,\n" +
                "`data`['tm_id'] as `tm_id`,\n" +
                "  CASE \n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) IS NULL THEN '未知'\n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) < 0 THEN '未知'\n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) <= 18 THEN '0-18'\n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) <= 25 THEN '18-25'\n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) <= 30 THEN '26-30'\n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) <= 35 THEN '31-35'\n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] AS DATE), CURRENT_DATE) <= 40 THEN '36-40'\n" +
                "        WHEN TIMESTAMPDIFF(YEAR, CAST(`data`['birthday'] At1S DATE), CURRENT_DATE) <= 50 THEN '41-50'\n" +
                "        ELSE '50以上'\n" +
                "    END as `age_group`,\n" +
                " ts\n" +
                "from topic_db\n" +
                "where `table` = 'user_info' and `type` = 'insert'");


//        //启动任务
//        env.execute();
    }
}
